# -*- coding: utf-8 -*-
from datetime import timedelta
from airflow.providers.apache.hdfs.sensors.web_hdfs import WebHdfsSensor


from utils.alerts.yl_threeSegCodeOnFailue import yl_threeSegCodeOnFailure
from utils.alerts.yl_threeSegCodeOnSuccess import yl_threeSegCodeOnSuccess


def kwargs():
    kwargs = {
        "db": "ai_group",
        "table": "geo_info_day",
        "desc": "爬虫经纬度geo数据",
        "taskid": "10160",
        "ifprivacy": 0,
        "warnignore": 0,
    }
    return kwargs

jms_aigroup__geo_info_day = WebHdfsSensor(
    pool='unlimited_pool',
    task_id="jms_aigroup__geo_info_day",
    ##filepath='/user/oozie-apps/hive/daily/done_flag/warehouse/ai_group/three_code/geo_info_day/{{ execution_date | cst_ds_nodash }}',
    filepath='/scheduler/ai_group/three_code/geo_info_day/{{ execution_date | cst_ds_nodash }}',
    execution_timeout=timedelta(hours=7),
    email=['yushuo@jtexpress.com','yl_bigdata@yl-scm.com'],
    # yarn_queue='pyspark',
    retries=2,
    on_success_callback=yl_threeSegCodeOnSuccess(kwargs(), dingding_conn_id="dingding_ThreeSeg_etl_info"),
    on_failure_callback=yl_threeSegCodeOnFailure(kwargs(), dingding_conn_id="dingding_ThreeSeg_etl_alert"),
)

# jms_aigroup__geo_info_day << jms_ai__sync_geo_info_day
